package com.idroc.rpc;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.mina.common.ByteBuffer;
import org.apache.mina.common.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.caucho.hessian.io.AbstractHessianInput;
import com.caucho.hessian.io.AbstractHessianOutput;
import com.caucho.hessian.io.Hessian2Input;
import com.caucho.hessian.io.Hessian2Output;
import com.caucho.hessian.io.HessianInput;
import com.caucho.hessian.io.HessianOutput;
import com.caucho.hessian.io.SerializerFactory;
import com.caucho.hessian.server.HessianSkeleton;
import com.idroc.message.AbstractMessage;
import com.idroc.message.Message;
import com.idroc.message.MessageHandler;
import com.idroc.message.MessageService;
import com.idroc.message.impl.MessageServer;

/**
 * rpc服务提供者
 * @author 赤脚詹(yuyoo4j@163.com)
 * @date Aug 17, 2009 4:22:18 PM 
 * @memo:
 */
public class RPCProvider {
	
	private static final Logger log = LoggerFactory.getLogger(RPCProvider.class);
	
	private MessageServer server = null;
	private RPCMessageHandler handler = null;
	
	private Map<Integer, Object> services = new ConcurrentHashMap<Integer, Object>();
	
	private SerializerFactory serializer = new SerializerFactory();
	
	private ExecutorService workpool = Executors.newFixedThreadPool(20);
	 
	RPCProvider (InetSocketAddress endpoint) {

		server = new MessageServer();
		Map config = new HashMap(); 
		config.put(MessageServer.MSG_SERVICE_HOST, endpoint.getHostName());
		config.put(MessageServer.MSG_SERVICE_PORT, Integer.toString(endpoint.getPort()));
		server.init(config);
		handler = new RPCMessageHandler();
		server.registerMsgHandler(RPCMessage.class, handler);
		server.start();
	}

	/**
	 * 对外暴露服务
	 * @param itfc -- 服务接口类
	 * @param impl -- 服务实现对象
	 */
	public void exportInterface(Class itfc, Object impl) {

		if (null != itfc) {
			HessianSkeleton proxy = new HessianSkeleton(impl, itfc);
			services.put(Integer.valueOf(itfc.getName().hashCode()), proxy);
		}
	}  
	
	/**
	 * 调用服务
	 * @param msg -- rpc消息
	 */
	private void invokeService(IoSession session, RPCMessage msg) {
		
		log.debug("开始rpc调用");
		try {
			
			int serviceId = msg.getServiceId();
			ByteBuffer data = msg.getRequestData();
			HessianSkeleton proxy = (HessianSkeleton) services.get(Integer.valueOf(serviceId));
			if (null == proxy) {
				throw new RuntimeException("不存在此服务");
			}
			ByteBuffer output = ByteBuffer.allocate(512);
			output.setAutoExpand(true);

		    AbstractHessianInput in = null;
		    AbstractHessianOutput out = null;
			int code = data.get();
		    int major = -1;
		    int minor = -1;
		    if (code == 'H') {
		    	major = data.get();
		    	minor = data.get();

		    	if (major != 0x02 || minor != 0x00) {
		    	  throw new IOException("Version " + major + "." + minor + " is not understood");
		    	}
		    	in = new Hessian2Input(data.asInputStream());
		    	out = new Hessian2Output(output.asOutputStream());

		    	in.readCall();
		    } else if (code == 'c') {
		    	major = data.get();
		    	minor = data.get();
		    	in = new HessianInput(data.asInputStream());
		    	if (major >= 2) {
		    		out = new Hessian2Output(output.asOutputStream());
		    	} else {
		    		out = new Hessian2Output(output.asOutputStream());
		    	} 
		    } else {
		    	throw new RuntimeException(" resouled data exception");
		    }
		    
		    log.debug("hessian message code:{}", code);
		    log.debug("major:{}, minor:{}", major, minor);
		     
		    in.setSerializerFactory(serializer);
		    out.setSerializerFactory(serializer);	
			proxy.invoke(in, out);
			out.close();	
			msg.setRequestData(output);
			session.write(msg);
			log.debug("结束rpc调用");
		} catch (Exception ex) { 
			log.error("rpc 调用异常", ex); 
		}
	}
	
	private void invoke(final IoSession session, final RPCMessage msg) {
		 
		Runnable task = new Runnable() {
			public void run() { 
				invokeService(session, msg);
			}			
		};
		workpool.execute(task);
		//invokeService(session, msg);
	}
	
	/**
	 * RPC message handler
	 * @author www.tianya.cn
	 * @creator 詹都林 (zhandl@hainan.net) 
	 * Aug 12, 2009 5:32:50 PM
	 * @memo:
	 */
	private class RPCMessageHandler implements MessageHandler {
		
		private final Logger log = LoggerFactory.getLogger(this.getClass());

		public void onAttached(MessageService service) { 
			
			log.debug("service:{}", service);
		}

		public void onMsgCast(IoSession session, Message message) { 
			log.debug("message:{}", message);
		}

		public void onMsgReceived(IoSession session, Message message) { 
			
			RPCMessage msg = (RPCMessage)message;
			log.debug("RPCMessage:{}", msg);
			RPCProvider.this.invoke(session, msg);
		}

		public void onMsgSent(IoSession session, Message message) { 
			log.debug("message:{}", message);
		}

		public void onReleased(MessageService service) { 
			log.debug("service:{}", service);
		} 
		
	}

}
